package com.leigang.rocketmq.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @author leig-a
 * @version 1.0
 * @desc
 * @date 2022/6/15 15:23
 */
@Component
@Slf4j
public class RocketProducer {
    @Value("${apache.rocketmq.producer.producerGroup}")
    private String producerGroup;

    @Value("${apache.rocketmq.producer.maxMessageSize}")
    private Integer maxMessageSize;

    @Value("${apache.rocketmq.producer.sendMsgTimeOut}")
    private Integer sendMsgTimeOut;

    @Value("${apache.rocketmq.producer.retryTimesWhenSendFailed}")
    private Integer retryTimesWhenSendFailed;

    private DefaultMQProducer producer;

    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;

    @PostConstruct
    public void defaultProducer() {
        producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(namesrvAddr);
        producer.setSendMsgTimeout(sendMsgTimeOut);
        producer.setMaxMessageSize(maxMessageSize);
        producer.setRetryTimesWhenSendFailed(retryTimesWhenSendFailed);
        producer.setVipChannelEnabled(false);
        try {
            producer.start();
            log.info("defaultProducer producer启动了");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }


    public String send(String topic,  String body) {
        String msgId = "";
        try {
            Message message = new Message(topic, body.getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult result = producer.send(message);
            log.info("send msgId: {}, send status:{}", result.getMsgId(), result.getSendStatus());
            msgId = result.getMsgId();
        } catch (Exception e) {
            msgId = "消息发送异常:" + e.getMessage();
            e.printStackTrace();
        }
        return "{\"MsgId\":\"" + msgId + "\"}";
    }


    public String send(String topic, String tags, String body) {
        String msgId = "";
        try {
            Message message = new Message(topic, tags, body.getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult result = producer.send(message);
            log.info("send msgId: {}, send status:{}", result.getMsgId(), result.getSendStatus());
            msgId = result.getMsgId();
        } catch (Exception e) {
            msgId = "消息发送异常:" + e.getMessage();
            e.printStackTrace();
        }
        return "{\"MsgId\":\"" + msgId + "\"}";
    }


    public String send(String topic, String tags, String key, String body) {
        String msgId = "";
        try {
            Message message = new Message(topic, tags, key, body.getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult result = producer.send(message);
            log.info("send msgId: {}, send status:{}", result.getMsgId(), result.getSendStatus());
            msgId = result.getMsgId();
        } catch (Exception e) {
            msgId = "消息发送异常:" + e.getMessage();
            e.printStackTrace();
        }
        return "{\"MsgId\":\"" + msgId + "\"}";
    }
}
